package com.example.klinemqflink.job;

import com.example.klinemqflink.fun.KLineAggregateFunction;
import com.example.klinemqflink.fun.SessionKLineAggregateFunction;
import com.example.klinemqflink.model.MarketData;
import com.example.klinemqflink.utils.Constant;
import com.example.klinemqflink.utils.LocalTimeUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
import java.util.concurrent.TimeUnit;

@Component
public class KlineCalculateService {
    private static final Logger logger = LoggerFactory.getLogger(KlineCalculateService.class);

    public void processLines(DataStream<MarketData> marketDataStream) {
        // Process for different intervals
        calculateKLines(marketDataStream, "1N", Time.minutes(1));
        calculateKLines(marketDataStream, "5N", Time.minutes(5));
        calculateKLines(marketDataStream, "15N", Time.minutes(15));
        calculateKLines(marketDataStream, "30N", Time.minutes(30));
        calculateKLines(marketDataStream, "1H", Time.hours(1));
    }

    private void calculateKLines(DataStream<MarketData> marketDataStream,
                                 String frequency, Time windowSize) {

        // 分配时间戳和水印
        marketDataStream = marketDataStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<MarketData>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<MarketData>() {
                            @Override
                            public long extractTimestamp(MarketData element, long recordTimestamp) {
//                                long timestampRounding = LocalTimeUtil.timestampRounding(element.getTimestamp());
//                                logger.info("水位时间,time={}",LocalTimeUtil.timestampToDateTime(timestampRounding));
                                return element.getTimestamp();
                            }
                        })
        );

        marketDataStream
//                .assignTimestampsAndWatermarks(new MarketDataWatermarkStrategy())
                .keyBy(data -> data.getChannel() + "_" + data.getInstrument() + "_" + data.getSymbol())
                .window(TumblingEventTimeWindows.of(windowSize))
                .allowedLateness(Duration.of(10, ChronoUnit.SECONDS))
                .sideOutputLateData(Constant.lateDataOutputTag)
                .aggregate(new KLineAggregateFunction(frequency))
                .addSink(new KlineSink());
    }
}
